package com.lefu.risk.storm.generdata;

import java.util.Map;
import java.util.Random;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class DateGenerateBolt extends BaseRichBolt{

	private static final long serialVersionUID = 1L;
	
	private static long id = 24874833;
	
	private OutputCollector collector;
	
	private String[] dates = null;
	
	private final static Random random = new Random();
	
	int count=0;
	
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
		dates= new String[30];
		for(int i= 0 ;i < 30;i++){
			dates[i] = "2016-6-"+(i+1);
			System.out.println(dates[i]);
		}
	}

	
	@Override
	public void execute(Tuple input) {
		String readLine = input.getStringByField("readLine");
		String[] splits = readLine.split("\t");
		splits[7]=splits[7].replace("2015-6-9", dates[random.nextInt(30)]);
		splits[0] = id++ +"";
		this.collector.emit(new Values(splits,splits[7].substring(0, 9).trim(),input.getLongByField("start")));
		
		if((++count % 5000)==0){
			System.out.println(splits[7].substring(0, 9).trim());
		}
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("arr","sort","start"));
	}

}
